package com.gome.ocean.server.service.impl;

import java.io.ByteArrayInputStream;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.instance.manager.model.Canal;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
import com.gome.framework.AppContext;
import com.gome.framework.Env;
import com.gome.framework.util.DateUtil;
import com.gome.ocean.common.enums.CangoRspStatus;
import com.gome.ocean.common.enums.CangoState;
import com.gome.ocean.common.enums.DataBaseType;
import com.gome.ocean.common.enums.RunMode;
import com.gome.ocean.common.enums.State;
import com.gome.ocean.dao.model.cango.CangoInstances;
import com.gome.ocean.dao.model.cango.CangoResponseVO;
import com.gome.ocean.dao.model.cango.CangoTable;
import com.gome.ocean.server.cache.CangoInstanceCache;
import com.gome.ocean.server.mysql.MysqlInstance;
import com.gome.ocean.server.mysql.MysqlParameter;
import com.gome.ocean.server.oracle.OracleInstance;
import com.gome.ocean.server.oracle.OracleInstancebak;
import com.gome.ocean.server.oracle.exception.YuGongException;
import com.gome.ocean.server.service.ICangoInstancesService;
import com.gome.ocean.server.service.ICangoManagerService;
import com.gome.ocean.server.service.ICangoTableService;
import com.gome.ocean.server.velocity.TemplateManager;
import com.gome.ocean.server.velocity.VelocityEngineManager;
import com.google.common.collect.Maps;

/**
 * @author zhangguoqing 2017年2月16日 上午9:26:12
 * @since 5.0.0
 */
@Service("iCangoManagerServiceImpl")
public class CangoManagerServiceImpl implements ICangoManagerService, InitializingBean, ApplicationListener {

    private static final Logger LOGGER                        = LoggerFactory.getLogger(CangoManagerServiceImpl.class);
    private static final String TEMPLATE_NAME                 = "yugong.vm";
    private static final String TEMPLATE_YUGONG_FILE_NAME_KEY = "yugong";

    @Autowired
    private ICangoInstancesService iCangoInstancesService;

    @Autowired
    private ICangoTableService iCangoTableService;

    private static final String CANAL_INSTANCE  = MysqlInstance.class.getSimpleName();
    private static final String ORACLE_INSTANCE = OracleInstancebak.class.getSimpleName();

    @Override
    public CangoResponseVO start(String instanceName) {
        return internalStart(instanceName);
    }

    /**
     * 启动实例.
     *
     * @param cangoRequestDto the canal req dto
     * @param startState 启动匹配类型的实例
     * @return the canal rsp dto
     * @date : 2016-07-05 13:58:14
     */
    private synchronized CangoResponseVO internalStart(String instanceName) {
        CangoResponseVO cangoResponseDto = new CangoResponseVO();

        // 检查是否已经添加过
        CangoInstances cangoInstancesPo = iCangoInstancesService.findByName(instanceName);
        if (cangoInstancesPo != null) {// 已添加过
            // 当前处于不可用状态
            if (CangoState.STOP == CangoState.valueOf(cangoInstancesPo.getState())) {
                CangoTable cangoTable = new CangoTable();
                cangoTable.setInstancesName(cangoInstancesPo.getName());
                List<CangoTable> pTableNames = iCangoTableService.getBeanListsByName(cangoTable.getInstancesName());
                if (DataBaseType.Oracle.getTypeName().equals(cangoInstancesPo.getTargetType().toString())) {
                    if (!CollectionUtils.isEmpty(pTableNames)) {
                        Set<String> tableNames = new HashSet<>(pTableNames.size());
                        for (CangoTable pTableName : pTableNames) {
                            tableNames.add(pTableName.getTableName());
                        }
                        cangoInstancesPo.setTableNames(tableNames);
                    }
                    startYuGong(cangoInstancesPo);
                } else {
                    startCanal(cangoInstancesPo);
                }

                cangoInstancesPo.setState(CangoState.START.getCode());
                cangoInstancesPo.setUpdateTime(new Date());
                iCangoInstancesService.update(cangoInstancesPo);
            }
            // 成功响应
            cangoResponseDto.setStatus(CangoRspStatus.SUCCESS);
            return cangoResponseDto;
        } else { // 未添加过
            cangoResponseDto.setStatus(CangoRspStatus.FAILURE);
            cangoResponseDto.setFailMsg(String.format("cannot find CangoInstance(%s)!", instanceName));
        }

        return cangoResponseDto;
    }

    @Override
    public CangoResponseVO add(CangoInstances instance) {
        CangoInstances cango = iCangoInstancesService.findByName(instance.getName());
        CangoResponseVO cangoResponseVo = new CangoResponseVO();
        cangoResponseVo.setStatus(CangoRspStatus.FAILURE);
        if (cango != null) {
            if (CangoState.DISABLE == CangoState.valueOf(cango.getState())) {
                cangoResponseVo.setFailMsg(String.format("CangoInstance(%s)has been disabled!", instance.getName()));
            } else if (CangoState.STOP == CangoState.valueOf(cango.getState())) {
                cangoResponseVo.setFailMsg(String.format("CangoInstance(%s)has been stop!", instance.getName()));
            } else if (CangoState.START == CangoState.valueOf(cango.getState())) {
                cangoResponseVo.setFailMsg(String
                    .format("CangoInstance(%s) is running!Don't repeat add the same of instance", instance.getName()));
            } else {
                cangoResponseVo.setFailMsg(String.format(
                    "CangoInstance(%s) is not anyway state!!!Please check this instance!!!", instance.getName()));
            }
            return cangoResponseVo;
        }
        return doAdd(instance);
    }

    /**
     * Do add cango response dto.
     *
     * @param cangoInstances the cango instances
     * @return the cango response dto
     */
    private synchronized CangoResponseVO doAdd(CangoInstances cangoInstances) {
        if (DataBaseType.Oracle.getTypeName().equals(cangoInstances.getTargetType().toString())) {
            // 启动yugong
            startYuGong(cangoInstances);
        } else {
            // 启动canal
            startCanal(cangoInstances);
        }
        // 持久化
        cangoInstances.setCreateTime(DateUtil.currDate());
        cangoInstances.setState(CangoState.START.getCode());
        iCangoInstancesService.saveOrUpdate(cangoInstances);
        return buildSuccessResponse();
    }

    /**
     * Start canal.
     *
     * @param cangoInstances the cango instances
     */
    private void startCanal(CangoInstances cangoInstances) {
        // 查看该数据库实例是否已启用
        String key = cangoInstances.getName();

        // 该数据库实例是否已经添加过
        MysqlInstance mysqlInstance = CangoInstanceCache.getMysqlInstance(key);
        if (mysqlInstance == null) {
            // 启动state为STOP的cangoInstance
            CangoInstances cangoInstancesPo = iCangoInstancesService.findByName(cangoInstances.getName());
            if (cangoInstancesPo != null) {
                if (CangoState.STOP == CangoState.valueOf(cangoInstancesPo.getState())) {
                    CangoTable cangoTable = new CangoTable();
                    cangoTable.setInstancesName(key);
                    List<CangoTable> cangoTablePos = iCangoTableService
                        .getBeanListsByName(cangoTable.getInstancesName());
                    mysqlInstance = buildMysqlInstance(cangoInstances);
                    // 将CanalInstance添加到本地缓存中
                    CangoInstanceCache.addMysqlInstance(key, mysqlInstance);
                    mysqlInstance.addTables(getStringTableName(cangoTablePos));
                }
            } else {
                mysqlInstance = buildMysqlInstance(cangoInstances);
                // 将CanalInstance添加到本地缓存中
                CangoInstanceCache.addMysqlInstance(key, mysqlInstance);
            }
        }

        // 校验注册的表是否合法
        Set<String> filteredTables = filterTableList(cangoInstances.getTableNames());
        // 获取尚未添加的表
        Set<String> newAddedTables = /*getNewAddedTables(filteredTables, mysqlInstance.getFullTableList())*/cangoInstances.getTableNames();

        mysqlInstance.addTables(filteredTables);
        // 持久化新增的表
        saveNewTables(newAddedTables, key);

        mysqlInstance.start();
    }

    /**
     * Build mysql instance mysql instance.
     *
     * @param cangoInstances the cango instances
     * @return the mysql instance
     */
    private MysqlInstance buildMysqlInstance(CangoInstances cangoInstances) {
        // 创建CanalInstance
        CanalParameter canalParameter = buildCanalParameter(cangoInstances);
        Canal canal = new Canal();
        canal.setName(cangoInstances.getName());
        canal.setCanalParameter(canalParameter);
        return new MysqlInstance(canal);
    }

    /**
     * Build canal parameter canal parameter.
     *
     * @param cangoInstances the canal instances
     * @return the canal parameter
     * @date :2016-04-26 17:20:18
     */
    private CanalParameter buildCanalParameter(CangoInstances cangoInstances) {
        MysqlParameter parameter = new MysqlParameter();
        parameter.setCanalName(cangoInstances.getName());
        parameter.setMetaMode(CanalParameter.MetaMode.ZOOKEEPER);
        parameter.setHaMode(CanalParameter.HAMode.HEARTBEAT);
        parameter.setIndexMode(CanalParameter.IndexMode.ZOOKEEPER);

        parameter.setMemoryStorageBufferSize(32 * 1024);

        parameter.setSourcingType(CanalParameter.SourcingType.MYSQL);
        parameter.setDbAddresses(Collections.singletonList(new InetSocketAddress(cangoInstances.getSourceHost(),
            cangoInstances.getSourcePort() < 0 ? cangoInstances.getSourcePort() : 3306)));
        parameter.setDbUsername(cangoInstances.getSourceUserName());
        parameter.setDbPassword(cangoInstances.getSourcePassword());

        parameter.setSlaveId(Long.parseLong(cangoInstances.getName().split("-")[0]));

        Env env = AppContext.getEnv();
        String zookUrl = env.get("zookeeperUrl");

        // zk设置
        parameter.setZkClusters(Arrays.asList(zookUrl.split(",")));

        parameter.setDefaultConnectionTimeoutInSeconds(30);
        parameter.setConnectionCharset("UTF-8");
        parameter.setConnectionCharsetNumber((byte) 33);
        parameter.setReceiveBufferSize(8 * 1024);
        parameter.setSendBufferSize(8 * 1024);

        // 过滤掉库
        // parameter.setBlackFilter("xx\\..*");
        // parameter.setBlackFilter(blackFilterConverter(cangoInstances.getBlackTables()));

        parameter.setDetectingEnable(false);
        parameter.setFilterTableError(true);
        parameter.setDetectingIntervalInSeconds(10);
        parameter.setDetectingRetryTimes(3);
        parameter.setDetectingSQL("SELECT 'x' FROM DUAL");

        return parameter;
    }

    /**
     * Gets new added tables.
     *
     * @param filteredTables the filtered tables
     * @param fullTableList the full table list
     * @return the new added tables
     */
    private Set<String> getNewAddedTables(Set<String> filteredTables, Set<String> fullTableList) {
        if (!CollectionUtils.isEmpty(filteredTables) && !CollectionUtils.isEmpty(fullTableList)) {
            // 解除引用共享
            HashSet<String> newAddedTables = new HashSet<>(filteredTables);
            for (String filteredTable : filteredTables) {
                for (String fullTable : fullTableList) {
                    // 考虑大小写
                    if (filteredTable.equalsIgnoreCase(fullTable)) {
                        newAddedTables.remove(filteredTable);
                    }
                }
            }
            return newAddedTables;
        }
        return null;
    }

    /**
     * Start yu gong.
     *
     * @param cangoInstances the canal instances
     * @date : 2016-07-05 14:05:30
     */
    private void startYuGong(CangoInstances cangoInstances) {
        // 查看该数据库实例是否已启用
        String key = cangoInstances.getName();
        // 该数据库实例是否已经添加过（相同的数据库主机、端口只需启动一个CanalInstance）
        OracleInstance cachedInstance = CangoInstanceCache.getOracleInstance(key);
        if (cachedInstance == null) {
            try {
                String instancesInfo = cangoInstances.getSourceHost() + ":" + cangoInstances.getSourcePort() + "/"
                                       + cangoInstances.getSourceDBname();
                LOGGER.info("start the OracleInstance ({}).", instancesInfo);
                if (StringUtils.isEmpty(cangoInstances.getSourceDBname())) {
                    throw new YuGongException("schema should not be null!");
                }

                doStartYugong(cangoInstances, key, instancesInfo);
                LOGGER.info("start the OracleInstance ({}) successfully!", instancesInfo);
            } catch (Throwable e) {
                CangoInstanceCache.removeOracleInstance(key);
                LOGGER.error("Something goes wrong when starting up the OracleInstance:\n{}",
                    ExceptionUtils.getFullStackTrace(e));
            }
        }
    }

    private void doStartYugong(CangoInstances cangoInstances, String key,
                               String instancesInfo) throws ConfigurationException, UnsupportedEncodingException {
        // 构造愚公配置文件信息
        PropertiesConfiguration config = new PropertiesConfiguration();
        Map<String, Object> objMap = Maps.newLinkedHashMap();
        String templeContent = TemplateManager.INSTANCE.getTemplate(TEMPLATE_NAME, TEMPLATE_YUGONG_FILE_NAME_KEY);
        String configDetailContent = VelocityEngineManager.INSTANCE.merge(templeContent,
            buildObjParam(objMap, cangoInstances));
        config.load(new ByteArrayInputStream(configDetailContent.getBytes("UTF-8")), "UTF-8");
        if (CangoState.STOP.getCode().equals(cangoInstances.getState())) {
            CangoTable cangoTable = new CangoTable().setInstancesName(key);
            List<CangoTable> cangoTables = iCangoTableService.getBeanListsByName(cangoTable.getInstancesName());
            cangoInstances.setTableNames(new HashSet(cangoTables));
            OracleInstance instance = new OracleInstance(cangoInstances.getName(), config);
            doStartYugong0(cangoInstances, key, instance, instancesInfo);
        }
    }

    public void doStartYugong0(CangoInstances cangoInstances, String key, OracleInstance instance,
                               String instancesInfo) {
        Set<String> filteredTables = filterTableList(cangoInstances.getTableNames());
        // 如果没有表，直接启动实例即可，且该实例的线程处于阻塞状态
        if (CollectionUtils.isEmpty(filteredTables)) {
            instance.start();
            CangoInstanceCache.addOracleInstance(key, instance);
            addShutDownHook(instance);
        } else {
            // List<String> addedTables = instance.addTableName(filteredTables);
            // if (!CollectionUtils.isEmpty(addedTables)) {
            // instance.start();
            // CangoInstanceCache.addOracleInstance(key, instance);
            // // 将成功添加到实例的表名持久化
            // saveNewTables(addedTables, key);
            // addShutDownHook(instance);
            // } else {
            // LOGGER.warn(
            // "OracleInstance ({}) execute addTableName method failed,
            // tableNames {}",
            // instancesInfo, String.valueOf(filteredTables));
            // }
        }
    }

    private void addShutDownHook(final OracleInstance instance) {
        Runtime.getRuntime().addShutdownHook(new Thread() {

            public void run() {
                if (instance.isStart()) {
                    try {
                        LOGGER.info("stop the OracleInstance");
                        instance.stop();
                    } catch (Throwable e) {
                        LOGGER.warn("something goes wrong when stopping OracleInstance:\n{}",
                            ExceptionUtils.getFullStackTrace(e));
                    } finally {
                        LOGGER.info("OracleInstance is down.");
                    }
                }
            }

        });
    }

    /**
     * 持久化instance中的表
     *
     * @param newAddedTables the new added tables
     * @param key the key
     */
    private void saveNewTables(Collection<String> newAddedTables, String key) {
        if (CollectionUtils.isEmpty(newAddedTables)) {
            List<CangoTable> cangoTables = new ArrayList<>();
            for (String tableName : newAddedTables) {
                CangoTable cangoTable = new CangoTable().setInstancesName(key).setTableName(tableName);
                cangoTables.add(cangoTable);
            }
            iCangoTableService.batchSave(cangoTables);
        }
    }

    private List<String> getStringTableName(List<CangoTable> cangoTablePos) {
        if (!CollectionUtils.isEmpty(cangoTablePos)) {
            List<String> tableNames = new ArrayList<>();
            for (CangoTable cangoTablePo : cangoTablePos) {
                tableNames.add(cangoTablePo.getTableName());
            }
            return tableNames;
        }
        return null;
    }

    /**
     * Filter table list set. note: 必须满足schema.table的格式
     *
     * @param tableNames the table names
     * @return the set
     */
    private Set<String> filterTableList(Set<String> tableNames) {
        if (!CollectionUtils.isEmpty(tableNames)) {
            Set<String> filteredTables = new HashSet<>();
            for (String tableName : tableNames) {
                if (StringUtils.isNotBlank(tableName) && tableName.contains(".") && !tableName.startsWith(".")
                    && !tableName.endsWith(".") && tableName.indexOf(".") == tableName.lastIndexOf(".")) {
                    filteredTables.add(tableName);
                }
            }

            return filteredTables;
        } else {
            return null;
        }
    }

    @Override
    public CangoResponseVO stop(String instanceName) {
        return internalStop(instanceName);
    }

    /**
     * 停用实例，并持久化实例状态.
     *
     * @param canalReqDto the canal req dto
     * @param stopState 需要持久化的状态
     * @return the canal rsp dto
     * @date : 2016-07-05 13:50:06
     */
    private synchronized CangoResponseVO internalStop(String instanceName) {
        CangoResponseVO cangoResponseDto = new CangoResponseVO();
        // 检查是否已经添加过
        CangoInstances cangoInstancesPo = iCangoInstancesService.findByName(instanceName);
        if (cangoInstancesPo != null) {// 已添加过
            // 当前处于可用状态
            if (CangoState.START == CangoState.valueOf(cangoInstancesPo.getState())) {
                cangoInstancesPo.setState(CangoState.STOP.getCode());
                cangoInstancesPo.setUpdateTime(new Date());
                iCangoInstancesService.update(cangoInstancesPo);

                if (DataBaseType.Oracle.getTypeName().equals(cangoInstancesPo.getTargetType().toString())) {
                    stopYuGong(cangoInstancesPo);
                } else {
                    stopCanal(cangoInstancesPo, false);
                }
            }

            // 成功响应
            cangoResponseDto.setStatus(CangoRspStatus.SUCCESS);
            return cangoResponseDto;
        } else { // 未添加过
            cangoResponseDto.setStatus(CangoRspStatus.FAILURE);
            cangoResponseDto.setFailMsg(String.format("cannot find CangoInstance(%s)!", instanceName));
        }
        return cangoResponseDto;
    }

    /**
     * Stop yu gong.
     *
     * @param cangoInstances the canal instances
     * @date : 2016-07-05 14:07:19
     */
    private void stopYuGong(CangoInstances cangoInstances) {
        String key = cangoInstances.getName();
        LOGGER.info("closing OracleInstance, name = {}, host = {}, dbName = {}",
            cangoInstances.getName(),
            cangoInstances.getSourceHost(),
            cangoInstances.getSourceDBname());
        OracleInstance instance = CangoInstanceCache.getOracleInstance(key);
        if (instance != null) {
            instance.stop();
            if (instance.isStop()) {
                CangoInstanceCache.removeOracleInstance(key);
            }
        } else {
            LOGGER.warn("OracleInstance close failed");
        }
    }

    /**
     * Stop canal.
     *
     * @param cangoInstances the canal instances
     * @param force 是否强制停用canal实例
     * @date :2016-04-25 16:04:01
     */
    private int stopCanal(CangoInstances cangoInstances, boolean force) {
        int countMysqlInstance = 0;
        // 查看该数据库实例是否已启用
        String key = cangoInstances.getName();
        // 该数据库实例是否已经添加过（相同的数据库主机、端口只需启动一个CanalInstance）
        MysqlInstance mysqlInstance = CangoInstanceCache.getMysqlInstance(key);
        if (mysqlInstance != null) {
            if (force) {
                Map<String, Object> map = Maps.newHashMap();
                map.put("host", cangoInstances.getSourceHost());
                map.put("port", cangoInstances.getSourcePort());
                map.put("state", CangoState.START.getCode());
                map.put("dbType", Integer.parseInt(DataBaseType.MySql.getTypeName()));
                // 在调用stopCanal之前，该canal实例已经持久化过
                List<CangoInstances> cangoInstancesList = iCangoInstancesService.findByCondition(map);
                if (!CollectionUtils.isEmpty(cangoInstancesList)) {
                    int stopState = cangoInstances.getState();
                    Date updateTime = new Date();

                    for (CangoInstances instances : cangoInstancesList) {
                        // 持久化
                        instances.setState(stopState);
                        instances.setUpdateTime(updateTime);
                        iCangoInstancesService.update(instances);
                        // stop
                        mysqlInstance.stop();
                        countMysqlInstance++;
                    }
                }
            }

            mysqlInstance.stop();
            countMysqlInstance++;

            if (!mysqlInstance.isStart()) {
                CangoInstanceCache.removeMysqlInstance(key);
            }
        }

        return countMysqlInstance;
    }

    @Override
    public CangoResponseVO enable(String instanceName) {
        return internalStart(instanceName);
    }

    @Override
    public CangoResponseVO disable(String instanceName) {
        return internalStop(instanceName);
    }

    @Override
    public CangoResponseVO check(String instanceName) {
        CangoResponseVO cangoResponseDto = new CangoResponseVO();
        // 检查是否已经添加过
        CangoInstances cangoInstancesPo = iCangoInstancesService.findByName(instanceName);
        if (cangoInstancesPo != null) { // 已添加过
            cangoResponseDto.setStatus(CangoRspStatus.SUCCESS);
            cangoResponseDto.setMsg(JSON.toJSONString(cangoInstancesPo));
            return cangoResponseDto;
        } else { // 未添加过
            cangoResponseDto.setStatus(CangoRspStatus.FAILURE);
            cangoResponseDto.setFailMsg(String.format("cannot find CangoInstance(%s)!", instanceName));
            return cangoResponseDto;
        }
    }

    @Override
    public synchronized CangoResponseVO startAll() {
        LOGGER.info("start all cangoInstance whose state is STOP!");
        Map<String, Object> map = Maps.newHashMap();
        map.put("state", State.FAIL.getValue());
        // 将STOP的全部实例启动！
        List<CangoInstances> instancesList = iCangoInstancesService.findByCondition(map);
        int countMysqlInstance = 0;
        int countOracleInstance = 0;
        if (!CollectionUtils.isEmpty(instancesList)) {
            Map<String, Integer> countMap = internalStartCanalAndYuGong(instancesList, null);
            countMysqlInstance = countMap.get(CANAL_INSTANCE);
            countOracleInstance = countMap.get(ORACLE_INSTANCE);

            // 批量更新
            iCangoInstancesService.batchUpdateState(instancesList);
        }
        LOGGER.info(
            "start all cangoInstance successfully whose state is STOP!，count MysqlInstance number {}, OracleInstance number {}!",
            countMysqlInstance,
            countOracleInstance);
        return buildSuccessResponse();
    }

    /**
     * 启动list中所有的canal和yugong.
     *
     * @param instancesList the instances list
     * @param logParseTime the logParseTime
     * @return the map
     * @date : 2016-07-05 16:39:37
     */
    private Map<String, Integer> internalStartCanalAndYuGong(List<CangoInstances> instancesList, Date logParseTime) {
        // canal实例数目
        int countMysqlInstance = 0;
        // yugong实例数目
        int countOracleInstance = 0;
        Map<String, Integer> countMap = new HashMap<>();

        for (CangoInstances cangoInstances : instancesList) {
            // 查看该数据库实例是否已启用
            if (cangoInstances.getTargetType().equals(Integer.parseInt(DataBaseType.Oracle.getTypeName()))) {
                startYuGong(cangoInstances);
                countOracleInstance++;
            } else {
                startCanal(cangoInstances);
                countMysqlInstance++;
            }

            // 重置为启用状态
            cangoInstances.setState(State.RUN.getValue());
        }

        countMap.put(CANAL_INSTANCE, countMysqlInstance);
        countMap.put(ORACLE_INSTANCE, countOracleInstance);
        return countMap;
    }

    @Override
    public synchronized CangoResponseVO stopAll() {
        Map<String, MysqlInstance> mysqlInstanceMap = CangoInstanceCache.getMysqlInstanceMap();
        Map<String, OracleInstance> oracleInstanceMap = CangoInstanceCache.getOracleInstanceMap();

        if (!CollectionUtils.isEmpty(mysqlInstanceMap) || !CollectionUtils.isEmpty(oracleInstanceMap)) {
            try {
                int canalInstanceSize = 0;
                int oracleInstanceSize = 0;
                if (!CollectionUtils.isEmpty(mysqlInstanceMap)) {
                    for (MysqlInstance mysqlInstance : mysqlInstanceMap.values()) {
                        CangoInstances cangoInstances = new CangoInstances();
                        cangoInstances.setName(mysqlInstance.getName());
                        CangoInstances cangoInstancesPo = iCangoInstancesService.findByName(cangoInstances.getName());
                        if (cangoInstancesPo != null) {
                            if (CangoState.START == CangoState.valueOf(cangoInstancesPo.getState())) {
                                cangoInstancesPo.setState(CangoState.STOP.getCode());
                                cangoInstancesPo.setUpdateTime(new Date());
                                iCangoInstancesService.update(cangoInstancesPo);
                                int count = stopCanal(cangoInstancesPo, true);
                                canalInstanceSize += count;
                            }
                        }
                    }

                    mysqlInstanceMap.clear();
                }

                if (!CollectionUtils.isEmpty(oracleInstanceMap)) {
                    for (OracleInstance oracleInstance : oracleInstanceMap.values()) {
                        CangoInstances cangoInstances = new CangoInstances();
                        cangoInstances.setName(oracleInstance.getName());
                        CangoInstances cangoInstancesPo = iCangoInstancesService.findByName(cangoInstances.getName());
                        if (cangoInstancesPo != null) {
                            // 当前处于可用状态
                            if (CangoState.START == CangoState.valueOf(cangoInstancesPo.getState())) {
                                cangoInstancesPo.setState(CangoState.STOP.getCode());
                                cangoInstancesPo.setUpdateTime(new Date());
                                iCangoInstancesService.update(cangoInstancesPo);
                                stopYuGong(cangoInstancesPo);
                                oracleInstanceSize++;
                            }
                        }
                    }

                    oracleInstanceMap.clear();
                }
                LOGGER.info("stopAll cangoInstance, count closed MysqlInstance number{}, OracleInstance number{}",
                    canalInstanceSize,
                    oracleInstanceSize);
            } catch (Exception e) {
                LOGGER.error("stopAll cangoInstance has an exception!", ExceptionUtils.getFullStackTrace(e.getCause()));
            }
        }
        return buildSuccessResponse();
    }

    @Override
    public synchronized CangoResponseVO shutdown() {
        stopAll();
        new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    Thread.currentThread().sleep(5000);
                    LOGGER.info("stop and exit cango successfully!");
                    System.exit(0);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();
        return buildSuccessResponse();
    }

    @Override
    public synchronized CangoResponseVO clearAll() {
        stopAll();
        clearAllInstance();
        LOGGER.info("clear all cangoInstance successfully");
        return buildSuccessResponse();
    }

    /**
     * 清空canal_instances表和canal_tablenames表.
     *
     * @date : 2016-07-19 14:12:18
     */
    private void clearAllInstance() {
        iCangoInstancesService.deleteAll();
        iCangoTableService.deleteAll();
    }

    @Override
    public void onApplicationEvent(ApplicationEvent applicationEvent) {
        // 容器启动完成后装载之前已添加的实例
        if (applicationEvent instanceof ContextRefreshedEvent) {
            LOGGER.info("spring container start all persisted cangoInstance!");
            Map<String, Object> map = Maps.newHashMap();
            map.put("state", State.RUN.getValue());
            List<CangoInstances> instancesList = iCangoInstancesService.findByCondition(map);
            int countMysqlInstance = 0;
            int countOracleInstance = 0;
            if (!CollectionUtils.isEmpty(instancesList)) {
                Map<String, Integer> countMap = internalStartCanalAndYuGong(instancesList, null);
                countMysqlInstance = countMap.get(CANAL_INSTANCE);
                countOracleInstance = countMap.get(ORACLE_INSTANCE);
            }

            LOGGER.info(
                "spring container start all persisted cangoInstance successfully, count MysqlInstance number {}, OracleInstance number {}!",
                countMysqlInstance,
                countOracleInstance);
        }

        if (applicationEvent instanceof ContextClosedEvent) {
            shutdown();
        }

    }

    private CangoResponseVO buildSuccessResponse() {
        return new CangoResponseVO().setStatus(CangoRspStatus.SUCCESS).setMsg("成功!");
    }

    private Map<String, Object> buildObjParam(Map<String, Object> objMap, CangoInstances cangoInstances) {
        objMap.put("sourceUserName", cangoInstances.getSourceUserName());
        objMap.put("sourcePassword", cangoInstances.getSourcePassword());
        objMap.put("sourceDBType",
            DataBaseType.getDataBaseType(cangoInstances.getSourceType()).getTypeFullName().toUpperCase());
        objMap.put("sourceJdbcUrl", cangoInstances.getSourceJDBCUrl());
        objMap.put("isRocketMqTarget", cangoInstances.getRocketMqVal());
        objMap.put("topicName", cangoInstances.getTopicname());
        objMap.put("nameSrvAddr", cangoInstances.getNamesrvaddr());
        objMap.put("targetJdbcUrl", cangoInstances.getTargetJDBCUrl());
        objMap.put("targetUserName", cangoInstances.getTargetUserName());
        objMap.put("targetPassword", cangoInstances.getTargetPassword());
        objMap.put("targetDBType",
            DataBaseType.getDataBaseType(cangoInstances.getTargetType()).getTypeFullName().toUpperCase());
        objMap.put("whiteTableName", cangoInstances.getTables());
        objMap.put("blackTableName",
            StringUtils.isNotBlank(cangoInstances.getBlackTables()) ? cangoInstances.getBlackTables() : "");
        objMap.put("moveMode", RunMode.getRunMode(cangoInstances.getMoveMode()).getValue());
        return objMap;
    }

    @Override
    public void afterPropertiesSet() throws Exception {

    }

    @Override
    public CangoResponseVO deleteByInstanceName(String instanceName) {
        CangoResponseVO responseVO = new CangoResponseVO();
        try {
            boolean flag = iCangoInstancesService.deleteByInstanceName(instanceName);
            if (flag) {
                responseVO.setMsg("Delete Success!");
                responseVO.setStatus(CangoRspStatus.SUCCESS);
            } else {
                responseVO.setFailMsg("Delete Failure!");
                responseVO.setStatus(CangoRspStatus.FAILURE);
            }
        } catch (Exception e) {
            LOGGER.error(ExceptionUtils.getMessage(e));
            responseVO.setFailMsg("Delete Error!" + ExceptionUtils.getMessage(e));
            responseVO.setStatus(CangoRspStatus.FAILURE);
        }
        return responseVO;
    }

    @Override
    public CangoResponseVO updateCangoInce(CangoInstances instances) {
        CangoResponseVO responseVO = new CangoResponseVO();
        try {
            boolean flag = iCangoInstancesService.update(instances);
            if (flag) {
                responseVO.setMsg("Update Success!");
                responseVO.setStatus(CangoRspStatus.SUCCESS);
            } else {
                responseVO.setFailMsg("Update Failure!");
                responseVO.setStatus(CangoRspStatus.FAILURE);
            }
        } catch (Exception e) {
            LOGGER.error(ExceptionUtils.getMessage(e));
            responseVO.setFailMsg("Update Error!" + ExceptionUtils.getMessage(e));
            responseVO.setStatus(CangoRspStatus.FAILURE);
        }
        return responseVO;
    }

}
